package com.deng.eventflow.core.tmf;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;

public class QueueSink extends Sink {
    protected Queue<Object>[] m_queues;

    public QueueSink(int in_arity)
    {
        super(in_arity);
        reset();
    }

    public QueueSink()
    {
        this(1);
    }

    @SuppressWarnings("unchecked")
    @Override
    public void reset()
    {
        super.reset();
        int arity = getInputArity();
        m_queues = new Queue[arity];
        for (int i = 0; i < arity; i++)
        {
            m_queues[i] = new ArrayDeque<Object>();
        }

    }

    @Override
    protected boolean compute(Object[] inputs, Queue<Object[]> outputs)
    {
        for (int i = 0; i < m_queues.length; i++)
        {
            Queue<Object> q = m_queues[i];
            if (inputs[i] != null)
            {
                q.add(inputs[i]);
            }
        }
        return true;
    }

    /**
     * Gets the queue corresponding to the <i>i</i>-th output of the sink
     *
     * @param i
     *          The position of the output. Must be non-negative and less than the
     *          queue's arity.
     * @return The queue
     */
    /* @requires i >= 0 && i < m_queues.length */
    public Queue<Object> getQueue(int i)
    {
        return m_queues[i];
    }

    /**
     * Gets the queue corresponding to the first output of the sink
     *
     * @return The queue
     */
    /* @requires m_queues.length > 0 */
    public Queue<Object> getQueue()
    {
        return getQueue(0);
    }

    /**
     * Removes the first event of all queues
     *
     * @return A vector containing the first event of all queues, or null
     */
    public Object[] remove()
    {
        Object[] out = new Object[m_queues.length];
        for (int i = 0; i < m_queues.length; i++)
        {
            Queue<Object> q = m_queues[i];
            if (q.isEmpty())
            {
                out[i] = null;
            }
            else
            {
                Object o = q.remove();
                out[i] = o;
            }
        }
        return out;
    }

    @Override
    public QueueSink duplicate(boolean with_state)
    {
        QueueSink qs = new QueueSink(getInputArity());
        if (with_state)
        {
            for (int i = 0; i < m_queues.length; i++)
            {
                qs.m_queues[i].addAll(m_queues[i]);
            }
        }
        return qs;
    }

    /**
     * @since 0.10.2
     */
    @Override
    public Object printState()
    {
        List<Queue<Object>> list = new ArrayList<Queue<Object>>(m_queues.length);
        for (Queue<Object> q : m_queues)
        {
            list.add(q);
        }
        return list;
    }

    /**
     * @since 0.10.2
     */
    @SuppressWarnings("unchecked")
    @Override
    public QueueSink readState(Object o)
    {
        List<Queue<Object>> list = (List<Queue<Object>>) o;
        QueueSink sink = new QueueSink(list.size());
        for (int i = 0; i < list.size(); i++)
        {
            sink.m_queues[i] = list.get(i);
        }
        return sink;
    }
}
